package com.allen.flink.dbus.function;

import com.alibaba.otter.canal.protocol.FlatMessage;
import com.allen.flink.dbus.enums.FlowStatusEnum;
import com.allen.flink.dbus.increments.IncrementSyncApp;
import com.allen.flink.dbus.model.Flow;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 处理方法
 *
 * KeyedBroadcastProcessFunction key类型、解析之后的输入类型、广播类型、输出
 */
public class DbusProcessFunction extends KeyedBroadcastProcessFunction<String, FlatMessage, Flow, Tuple2<FlatMessage, Flow>> {
    /**
     * 处理数据流方法
     * @param value kafka的消息
     * @param ctx 只读上下文
     * @param out 输出
     * @throws Exception
     */
        public void processElement(FlatMessage value, ReadOnlyContext ctx, Collector<Tuple2<FlatMessage, Flow>> out) throws Exception {
            //获取配置流
            Flow flow = ctx.getBroadcastState(IncrementSyncApp.flowStateDescriptor).get(value.getDatabase() + value.getTable());
        
            if(null != flow && flow.getStatus() == FlowStatusEnum.FLOWSTATUS_RUNNING.getCode()){
                out.collect(Tuple2.of(value,flow));
            }
    }
    
    /**
     * 广播变量处理方法
     * @param flow 广播变量
     * @param ctx 上下文
     * @param out 输出
     * @throws Exception
     */
    public void processBroadcastElement(Flow flow, Context ctx, Collector<Tuple2<FlatMessage, Flow>> out) throws Exception {

        //获取state状态
        BroadcastState<String, Flow> broadcastState = ctx.getBroadcastState(IncrementSyncApp.flowStateDescriptor);

        //更新state
        broadcastState.put(flow.getDatabaseName()+flow.getTableName(),flow);
    }
}
